﻿// *----------------------------------------------------------------
// Copyright (C) 2017 通通优品
// 版权所有。
// 
// 类名：RabbitMqMessageSubscriber.cs
// 功能描述：TongTongMall.RabbitMq
// 
// Create User：jym 2017/02/09
// 
// Edit User：UserName,EditTime
// Describe：Describe
// ----------------------------------------------------------------*/
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Configuration;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using TongTongMall.Commanding.Messaging;
using TongTongMall.Events;
using TongTongMall.Extensions;
using TongTongMall.Logging;
using TongTongMall.UI;
using TongTongMall.Util.Etc;

namespace TongTongMall.RabbitMq
{
    /// <summary>
    /// 订阅消息
    /// 消息的消费者
    /// </summary>
    public class RabbitMqMessageSubscriber : DisposableObject, IMessageSubscriber
    {
        private readonly string exchangeName;
        private readonly string queueName;
        private readonly IConnection connection;
        private readonly IModel channel;
        private bool disposed;

        public event EventHandler<MessageReceivedEventArgs> MessageReceived;

        /// <summary>
        /// 初始化rabbitmq 订阅消息模式
        /// </summary>
        /// <param name="uri"></param>
        public RabbitMqMessageSubscriber(string uri, string exchangeName, string queueName)
        {
            this.exchangeName = exchangeName;
            this.queueName = queueName;
            var factory = new ConnectionFactory() {
                Uri = uri,
                //自动重连
                AutomaticRecoveryEnabled =true,
                //心跳时间
                RequestedHeartbeat =60
            };
            this.connection = factory.CreateConnection();
            this.channel = connection.CreateModel();
        }

        public void Subscribe()
        {
            //不启用路由认证，有rabbitmq直接进行广播转发
            channel.ExchangeDeclare(exchange: this.exchangeName, type: "fanout");
            //每次只消费一个队列
            channel.BasicQos(0, 1, false);
            //指明队列内容
            channel.QueueDeclare(queue: this.queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
            //指明绑定在队列内，每个元素的内容
            channel.QueueBind(queue: this.queueName,
                              exchange: this.exchangeName,
                              routingKey: "");
            //被动订阅模式
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                try
                {
                    var body = e.Body;
                    var json = Encoding.UTF8.GetString(body);
                    var message = JsonConvert.DeserializeObject<MessageData>(json);
                    this.OnMessageReceived(new MessageReceivedEventArgs(message));
                    //是否回发rabbitmq 消息，rabbitmq自动对队列进行管理
                    channel.BasicAck(e.DeliveryTag, multiple: false);
                }
                catch (UserFriendlyException ex)
                {
                    LogHelper.LogException(ex);
                }
                catch (Exception ex)
                {
                    //程序异常 ，需要重新进行消费
                    channel.BasicReject(e.DeliveryTag, true);

                    LogHelper.LogException(ex);
                }
            };
            
            //接受消息
            channel.BasicConsume(queue: queueName,
                                 noAck: false,
                                 consumer: consumer);
        }

        /// <summary>
        /// 消息接受方
        /// </summary>
        /// <param name="e"></param>
        private void OnMessageReceived(MessageReceivedEventArgs e)
        {
            this.MessageReceived?.Invoke(this, e);
        }

        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (!disposed)
                {
                    this.channel.Dispose();
                    this.connection.Dispose();
                    disposed = true;
                }
            }
        }
    }
}
